package com.cellulam.framework.core.async;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.concurrent.*;

/**
 * 基于内存的延迟队列，仅用于要求不高的场景，持久化执行任务使用TaskSchedule
 */
@Slf4j
@Component
public class DelayExecutor {

    private DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();

    private final ExecutorService POOL = Executors.newCachedThreadPool();

    @PostConstruct
    private void postConstruct() {
        this.startDelayConsumer();
    }

    public void addTask(Executor executor, int delayMills) {
        this.delayQueue.add(new DelayedTask(executor, delayMills));
    }

    public void addTask(Executor executor, LocalDateTime exceptedStartTime) {
        this.delayQueue.add(new DelayedTask(executor, exceptedStartTime));
    }

    private void startDelayConsumer() {
        new Thread(() -> {
            while (true) {
                try {
                    DelayedTask item = this.delayQueue.take();
                    //避免阻塞延迟队列，使用单独线程池执行
                    POOL.execute(() -> item.execute());
                } catch (Exception e) {
                    log.error("消费队列异常", e);
                }
            }
        }).start();
    }

    private static class DelayedTask implements Delayed {

        private long expectedBeginTime;

        private Executor executor;

        public DelayedTask(Executor executor, int delayMills) {
            this.init(executor, delayMills);
        }

        public DelayedTask(Executor executor, LocalDateTime exceptedStartTime) {
            this.init(executor, exceptedStartTime.atZone(ZoneId.systemDefault())
                    .toInstant()
                    .toEpochMilli());
        }

        private void init(Executor executor, int delayMills) {
            this.init(executor, delayMills + System.currentTimeMillis());
        }

        private void init(Executor executor, long expectedBeginTime) {
            this.executor = executor;
            this.expectedBeginTime = expectedBeginTime;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            long diff = expectedBeginTime - System.currentTimeMillis();
            long convert = unit.convert(diff, TimeUnit.MILLISECONDS);
            log.debug("getDelay convert is {}", convert);
            return convert;
        }

        @Override
        public int compareTo(Delayed o) {
            return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
        }

        public void execute() {
            try {
                this.executor.execute();
            } catch (Exception e) {
                log.error("消息队列异常");
            }
        }
    }
}
